package com.atguigu.chapter07.c_timer;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @ClassName: Flink14_Timeer_Process
 * @Description:
 * @Author: kele
 * @Date: 2021/4/7 13:16
 *
 * 处理时间的定时器
 **/
public class Flink14_Timeer_Process {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.socketTextStream("hadoop162",8888)
                .map(datas -> {

                    String[] data = datas.split(",");
                    return new WaterSensor(data[0],
                            Long.parseLong(data[1]) * 1000,
                            Integer.valueOf(data[2]));
                })
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                        long timeMillis = System.currentTimeMillis();
                        //注册定时器
                        ctx.timerService().registerProcessingTimeTimer(timeMillis+5000);
                        out.collect("注册："+value.getId()+",注册时间为："+timeMillis);
                    }

                    //定时器时间到处理
                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect("触发时间为："+timestamp);
                    }
                }).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
